package org.apache.ocean.merge;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Logger;

import org.apache.lucene.index.Term;
import org.apache.ocean.CLong;
import org.apache.ocean.CSortField;
import org.apache.ocean.CSortFields;
import org.apache.ocean.ClusterConstants;
import org.apache.ocean.CurrentIndexSchema;
import org.apache.ocean.HttpParameters;
import org.apache.ocean.IndexSchema;
import org.apache.ocean.OceanDocument;
import org.apache.ocean.OceanDocumentComparator;
import org.apache.ocean.SearchRequest;
import org.apache.ocean.SearchResponse;
import org.apache.ocean.TermFrequencies;
import org.apache.ocean.TimeoutState;
import org.apache.ocean.XMLUtil;
import org.apache.ocean.OceanXML.RequestException;
import org.apache.ocean.name.Node;
import org.apache.ocean.nio.OceanNioHttpClient;
import org.apache.ocean.nio.OceanNioHttpClient.OceanResponse;
import org.apache.ocean.nio.OceanNioHttpClient.RequestID;
import org.apache.ocean.util.CircularQueue;
import org.jdom.Element;

public class MergeClient {
	public static Logger log = Logger.getLogger(MergeClient.class.getName());
	private OceanNioHttpClient oceanNioHttpClient;
	private MergeClientInterface mergeClientInterface;

	public MergeClient(MergeClientInterface mergeClientInterface, OceanNioHttpClient oceanNioHttpClient) {
		this.mergeClientInterface = mergeClientInterface;
		this.oceanNioHttpClient = oceanNioHttpClient;
	}

	// TODO: how many errors per time means a node is unavailable
	public static interface MergeClientInterface {
		public Map<String, List<Node>> getReadableCellNodes(TimeoutState timeoutState) throws Exception;
		public List<Node> getAddDocumentNodes(TimeoutState timeoutState) throws Exception;
		public Node getUpdateDocumentNode(String shardID, TimeoutState timeoutState) throws Exception;
		public void reportReadFailure(String nodeID);
		public CurrentIndexSchema getCurrentIndexSchema();
	}

	public Map<String, OceanResponse> request(MultiCellRequest multiCellRequest) throws Exception {
		for (int x = 0; x < multiCellRequest.retries; x++) {
			try {
				Map<String, OceanResponse> responses = oceanNioHttpClient.requestOceanXMLs(multiCellRequest.getRequestIDs(), multiCellRequest.httpParameters,
						multiCellRequest.timeoutState);
				return responses;
			} catch (RequestException requestException) {
				log.severe("error requesting " + requestException.url + " attachment: " + requestException.attachment);
				Node node = (Node) requestException.attachment;
				multiCellRequest.iterate(node.assignedCellID);
			}
		}
		throw new Exception("unable to obtain complete responses from all cells");
	}

	public static class MultiCellRequest {
		public Map<String, CellSession> cellMap = new HashMap<String, CellSession>();
		TimeoutState timeoutState;
		HttpParameters httpParameters;
		String requestBody;
		String method;
		int retries;

		public MultiCellRequest(int retries, HttpParameters httpParameters, TimeoutState timeoutState) {
			this.retries = retries;
			this.httpParameters = httpParameters;
			this.timeoutState = timeoutState;
			method = "get";
		}

		public List<RequestID> getRequestIDs() {
			List<RequestID> requestIDs = new ArrayList<RequestID>();
			for (CellSession cellSession : cellMap.values()) {
				Node node = cellSession.get();
				requestIDs.add(new RequestID(node, node.url));
			}
			return requestIDs;
		}

		public void iterate(String cellID) {
			cellMap.get(cellID).next();
		}

		public void addCell(String cellID, List<Node> nodes) {
			cellMap.put(cellID, new CellSession(nodes));
		}
	}

	public static class CellSession {
		public CircularQueue<Node> nodeQueue;

		public CellSession(Collection<Node> nodes) {
			nodeQueue = new CircularQueue<Node>(nodes);
		}

		public void next() {
			nodeQueue.poll();
		}

		public Node get() {
			return nodeQueue.peek();
		}
	}

	// TODO: capture timeout and io exception in niohttpclient and send back in
	// requestexception
	public MultiCellRequest createAllCellRequest(HttpParameters httpParameters, TimeoutState timeoutState) throws Exception {
		MultiCellRequest multiCellRequest = new MultiCellRequest(4, httpParameters, timeoutState);
		Map<String, List<Node>> map = mergeClientInterface.getReadableCellNodes(timeoutState);
		for (Map.Entry<String, List<Node>> entry : map.entrySet()) {
			multiCellRequest.addCell(entry.getKey(), entry.getValue());
		}
		return multiCellRequest;
	}

	public SearchResponse search(SearchRequest searchRequest, TimeoutState timeoutState) throws Exception {
		IndexSchema indexSchema = mergeClientInterface.getCurrentIndexSchema().getIndexSchema(timeoutState);
		HttpParameters parameters = searchRequest.toHttpParameters();
		MultiCellRequest multiCellRequest = createAllCellRequest(parameters, timeoutState);
		Map<String,OceanResponse> responseMap = request(multiCellRequest);
		SearchResponse searchResponse = new SearchResponse();
		CSortFields sortFields = null;
		if (searchRequest.sortFields == null) {
			sortFields = new CSortFields();
			sortFields.add(new CSortField(ClusterConstants.SCORE_FIELD, true));
		} else {
			sortFields = searchRequest.sortFields;
		}
		OceanDocumentComparator comparator = new OceanDocumentComparator(sortFields, OceanDocumentComparator.NULLS_ARE_LEAST, indexSchema);
		List<OceanDocument> totalOceanDocuments = new ArrayList<OceanDocument>();
		CLong totalNumFound = new CLong(0);
		for (OceanResponse oceanResponse : responseMap.values()) {
			SearchResponse nodeSearchResponse = new SearchResponse(oceanResponse.oceanXML);
			totalNumFound.value = totalNumFound.value + nodeSearchResponse.numFound.value;
			totalOceanDocuments.addAll(nodeSearchResponse.oceanDocuments);
		}
		Collections.sort(totalOceanDocuments, comparator);
		List<OceanDocument> oceanDocuments = slice(totalOceanDocuments, (int)searchRequest.rows.value);
		searchResponse.oceanDocuments = oceanDocuments;
		return searchResponse;
	}

	private List<OceanDocument> slice(List<OceanDocument> totalOceanDocuments, int max) {
		int count = 0;
		List<OceanDocument> oceanDocuments = new ArrayList<OceanDocument>(max);
		for (OceanDocument doc : totalOceanDocuments) {
			count++;
			if (count == oceanDocuments.size()) break;
			oceanDocuments.add(doc);
		}
		return oceanDocuments;
	}
	
	public TermFrequencies getTermFrequencies(String queryString, TimeoutState timeoutState) throws RequestException, Exception {
		HttpParameters parameters = new HttpParameters("action", "gettermfrequencies", "q", queryString);
		MultiCellRequest multiCellRequest = createAllCellRequest(parameters, timeoutState);
		Map<String, OceanResponse> responseMap = request(multiCellRequest);
		TermFrequencies termFrequencies = new TermFrequencies();
		for (OceanResponse oceanResponse : responseMap.values()) {
			List<Element> elements = XMLUtil.getChildren(oceanResponse.oceanXML.bodyElement);
			TermFrequencies responseTermFrequencies = new TermFrequencies(elements);
			for (Map.Entry<Term, AtomicLong> entry : responseTermFrequencies.entrySet()) {
				termFrequencies.add(entry.getKey(), entry.getValue());
			}
		}
		return termFrequencies;
	}
}
